package org.project.order

import org.FlinkStreamApp
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector
import org.project.bean.OrderEvent

import scala.collection.Map

/**
 * description ：订单超时 cep 实现
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 13:56
 * modified By ：
 * version:    : 1.0
 */
object OrderTimeoutWithCep extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 创建订单数据
    val source = environment.fromElements(
      OrderEvent("1", "create", "1558430842"),
      OrderEvent("2", "create", "1558430843"),
      OrderEvent("2", "pay", "1558430844"),
      OrderEvent("3", "pay", "1558430942"),
      OrderEvent("4", "pay", "1558430943")
    ).assignAscendingTimestamps(_.eventTime.toLong * 1000)

    // cep 实现超时提醒
    val pattern = Pattern
      .begin[OrderEvent]("begin").where(_.eventType == "create")
      .next("next").where(_.eventType == "pay")
      .within(Time.seconds(5))


    // pattern 匹配
    val selectSteam = CEP.pattern(source.keyBy(_.orderId), pattern)

    // 侧输出标签
    val outputTag = new OutputTag[String]("selectStream")

    // 匹配超时的事件
    val patternFlatTimeoutFunction = (map: Map[String, Iterable[OrderEvent]], timestamp: Long, out: Collector[String]) => {
      val orderStart = map("begin")
      for (order <- orderStart) {
        out.collect("订单ID为 " + order.orderId + " 在5s以内没有支付")
      }
    }

    // 匹配正常的事件
    val patternFlatSelectFunction = (map: Map[String, Iterable[OrderEvent]], out: Collector[String]) => {
      val orderStart = map("begin")
      for (order <- orderStart) {
        out.collect("订单ID为 " + order.orderId + " 在5s以内完成了支付")
      }
    }

    // pattern 选择
    // outputTag 这个是超时事件输出标签，而正常事件就往流继续走
    val flatSelectResult = selectSteam.flatSelect(outputTag)(patternFlatTimeoutFunction)(patternFlatSelectFunction)


    // 超时事件输出
    flatSelectResult.getSideOutput(outputTag).print()

    // 正常事件输出
    flatSelectResult.print()
  }
}
